package simpleTaskDispatcher.model;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Semaphore;
import java.util.logging.Logger;

import simpleTaskDispatcher.model.enums.DispatchingEvent;
import simpleTaskDispatcher.model.enums.Priority;
import simpleTaskDispatcher.model.enums.TaskPlacement;
import simpleTaskDispatcher.model.tasks.DelayedTasks;
import simpleTaskDispatcher.model.tasks.ITask;
import simpleTaskDispatcher.model.tasks.ITaskAllowedAfterTaskQueueClose;
import simpleTaskDispatcher.model.tasks.impl.BarrierTask;
import simpleTaskDispatcher.model.tasks.impl.ReleaseWorkerTask;
import simpleTaskDispatcher.model.tasks.impl.SetMaximalWorkersTask;
import simpleTaskDispatcher.model.tasks.impl.SetMinimalWorkersTask;
import simpleTaskDispatcher.model.tasks.impl.SetOptimalWorkersTask;
import simpleTaskDispatcher.model.tasks.impl.SetParameterTask;

public class TaskQueue {

	volatile int plannedNbWorkers = 0;
	
	private int minimalWorkers;
	private Integer optimalWorkers;
	private int maximalWorkers;
	
	private double realisedWorkValue;
	private double currentTotalWorkValue;
	private Double estimatedWorkValue;
	
	private boolean closeSignaled;
	
	private Map<String, Object> sharedParameters = new HashMap<String, Object>();
	
	private BlockingDeque<ITask> tasks = new LinkedBlockingDeque<ITask>();
	
	private final String name;
	
	private WorkersPool pool;
	
	private Semaphore takingTask = new Semaphore(1);
	
	private Semaphore leaveSemaphore = new Semaphore(0);
	
	private Date creation;
	
	private Map<String, List<DelayedTasks>> tasksWithDependencies = new HashMap<String, List<DelayedTasks>>();
	
	private Semaphore modifyingSharedParameter = new Semaphore(1);
	
	//Lesser means more important
	private int priority;
	
	//This is not the same thing as nbWorkers. If one workers is assigned and terminated 5 times, this property will be 5
	private int assignedTaskQueueWorkers;
	private Semaphore terminatedTaskQueueWorkers = new Semaphore(0);
	
	private Logger LOG;
	
	protected void setPool(WorkersPool pool) {
		this.pool = pool;
	}

	public TaskQueue(String name, int minimalWorkers, int optimalWorkers, int maximalWorkers) {
		this.minimalWorkers = minimalWorkers;
		this.optimalWorkers = optimalWorkers;
		this.maximalWorkers = maximalWorkers;
		this.name = name;
		this.creation = new Date();
		LOG = Logger.getLogger("Task queue " + name);
	}

	public TaskQueue(String name, int minimalWorkers, int maximalWorkers) {
		this.minimalWorkers = minimalWorkers;
		this.optimalWorkers = null;
		this.maximalWorkers = maximalWorkers;
		this.name = name;
		this.creation = new Date();
		LOG = Logger.getLogger("Task queue " + name);
	}
	
	public TaskQueue(String name) {
		this.minimalWorkers = 1;
		this.optimalWorkers = null;
		this.maximalWorkers = 10000;
		this.name = name;
		this.creation = new Date();
		LOG = Logger.getLogger("Task queue " + name);
	}
	
	protected void assignate(List<Worker> workers) {
		
		for(Worker worker : workers) {
			if (!worker.isAlive()) {
				throw new RuntimeException("Worker must be alived");
			}
			assignedTaskQueueWorkers += 1;
			worker.assignToTaskQueue(this);
			incrementNbWorkers();
		}
	}
	
	public boolean isAssignable(int newWorkers) {
		return plannedNbWorkers + newWorkers <= maximalWorkers;
	}
	
	public final ITask take() throws InterruptedException {
		takingTask.acquire();
		
		
		ITask task= tasks.takeFirst();
		
		if (task instanceof ReleaseWorkerTask) {
			ReleaseWorkerTask releaseTask = (ReleaseWorkerTask) task;
			if (releaseTask.isReleaseAll()) {
				
				decrementNbWorkers();
				tasks.put(releaseTask);
			}
			
		} else if (task instanceof BarrierTask) {
			BarrierTask barrierTask = (BarrierTask) task;
			if (plannedNbWorkers == barrierTask.getCurrentHaltedWorkers() + 1) {
				barrierTask.getWorkersWaitingForBarrier().release(plannedNbWorkers);
				barrierTask.getBarrierFinished().release();
			} else {
				BarrierTask otherBarrierTask = new BarrierTask(barrierTask.getWorkersWaitingForBarrier(), barrierTask.getBarrierFinished(), barrierTask.getCurrentHaltedWorkers() + 1);
				putInPriority(otherBarrierTask);
			}
			
		} else if (task instanceof SetMinimalWorkersTask) {
			apply((SetMinimalWorkersTask) task);
			
		} else if (task instanceof SetOptimalWorkersTask) {
			apply((SetOptimalWorkersTask) task);
			
		} else if (task instanceof SetMaximalWorkersTask) {
			apply((SetMaximalWorkersTask) task);
		}
		
		takingTask.release();
		//LOG.info(task.toString());
		return task;
	}
	
	private void apply(SetMinimalWorkersTask task) throws InterruptedException {
		int newMinimum = task.getMinimalWorkers();
		synchronized (pool) {
			LOG.info("Old minimum: " + this.minimalWorkers + "; New minimum: " + newMinimum);
			this.minimalWorkers = ((SetMinimalWorkersTask) task).getMinimalWorkers();
			if (this.optimalWorkers != null && this.optimalWorkers < this.minimalWorkers) {
				this.optimalWorkers = this.minimalWorkers;
			}
			pool.doDispatching(DispatchingEvent.MINIMAL_WORKERS_CHANGED);
		}
	}
	
	private void apply(SetOptimalWorkersTask task) throws InterruptedException  {
		
		int newOptimimal = task.getOptimalWorkers();
		
		synchronized (pool) { 
			LOG.info("Old optimal: " + this.optimalWorkers + "; New optimal: " + newOptimimal);
			this.optimalWorkers = (task).getOptimalWorkers();
			if (this.optimalWorkers != null && (this.optimalWorkers < this.minimalWorkers || this.optimalWorkers > this.maximalWorkers)) {
				this.optimalWorkers = null;
			}
			pool.doDispatching(DispatchingEvent.OPTIMAL_WORKERS_CHANGED);
		}
	}
	
	private void apply(SetMaximalWorkersTask task) throws InterruptedException {
		int newMax = (task).getMaximalWorkers();
		
		synchronized (pool) {
			LOG.info("Old maximum: " + this.maximalWorkers + "; New maximum: " + newMax);
			this.maximalWorkers = newMax;
			if (this.optimalWorkers != null && this.optimalWorkers > this.maximalWorkers) {
				this.optimalWorkers = this.maximalWorkers;
			}
			
			if (plannedNbWorkers > maximalWorkers) {
				int releasedWorkers = plannedNbWorkers - maximalWorkers;
				releaseWorkers(releasedWorkers, false, TaskPlacement.PLACE_AT_START_OF_QUEUE, DispatchingEvent.MAXIMAL_WORKERS_CHANGED);
			} else {
				pool.doDispatching(DispatchingEvent.MAXIMAL_WORKERS_CHANGED);
			}
		}
		

	}
	
	public boolean isCloseSignaled() {
		return closeSignaled;
	}

	synchronized void finished(ITask task) {
		double workValue = task.getWorkValue();
		if (workValue > 0) {
			realisedWorkValue += workValue;
		}
	}
	
	public synchronized double getTotalWorkValue() {
		if (closeSignaled || estimatedWorkValue == null) {
			return currentTotalWorkValue;
		} else {
			return Math.max(estimatedWorkValue, currentTotalWorkValue);
		}
	}
	
	public synchronized float getProgression() {
		double total = getTotalWorkValue();
		return total == 0 ? 0 : (float) (realisedWorkValue / total);
	}
	
	public synchronized TaskQueue setEstimatedWorkValue(Double estimatedWorkValue) {
		this.estimatedWorkValue = estimatedWorkValue;
		return this;
	}

	public synchronized final void putInPriority(ITask newTask) throws InterruptedException {
		put(newTask, TaskPlacement.PLACE_AT_START_OF_QUEUE);
	}
	
	public synchronized final void putAllInPriority(List<ITask> newTasks) throws InterruptedException {
		putAll(newTasks, TaskPlacement.PLACE_AT_START_OF_QUEUE, false);
	}
	
	private synchronized void put(ITask task, TaskPlacement where) throws InterruptedException {
		if (closeSignaled &&!(task instanceof ITaskAllowedAfterTaskQueueClose)) {
			throw new IllegalStateException("This task queue has been closed, you cannot submit this task");
		}
		if (where == TaskPlacement.PLACE_AT_START_OF_QUEUE || closeSignaled) {
			tasks.putFirst(task);
		} else {
			tasks.putLast(task);
		}
	}
	
	private synchronized void putAll(List<ITask> tasks, TaskPlacement where, boolean allowedOnTaskQueueSignaledToClose) throws InterruptedException {
		//Add all of them or nothing
		if (!allowedOnTaskQueueSignaledToClose) {
			for(ITask task : tasks) {
				if (closeSignaled &&!(task instanceof ITaskAllowedAfterTaskQueueClose)) {
					throw new IllegalStateException("This task queue has been closed, you cannot submit this task");
				}
			}
		}
		
		for(int i = 0; i < tasks.size(); i++) {
			if (where == TaskPlacement.PLACE_AT_START_OF_QUEUE || closeSignaled) {
					this.tasks.putFirst(tasks.get(tasks.size() - i - 1));
			} else {
					this.tasks.putLast(tasks.get(i));
			}
		}
	}

	public synchronized final void put(ITask newTask) throws InterruptedException {
		put(newTask, TaskPlacement.PLACE_AT_END_OF_QUEUE);
		double workValue = newTask.getWorkValue();
		if (workValue > 0) {
			currentTotalWorkValue += workValue;
		}
	}
	
	public synchronized final void putAll(List<ITask> newTasks) throws InterruptedException {
		putAll(newTasks, TaskPlacement.PLACE_AT_END_OF_QUEUE, false);
	}
	
	public synchronized final void put(DelayedTasks delayedTasks) throws InterruptedException {
		for(ITask afterTask : delayedTasks.getDependencies()) {
			if (afterTask.isExecuted()) {
				delayedTasks.decrementDependencies();
			} else {
				
				List<DelayedTasks> tasksDepending = tasksWithDependencies.get(afterTask.getUniqueID());
				if (tasksDepending == null) {
					tasksDepending = new ArrayList<DelayedTasks>();
					tasksDepending.add(delayedTasks);
					tasksWithDependencies.put(afterTask.getUniqueID(), tasksDepending);
				} else {
					tasksDepending.add(delayedTasks);
				}
			}
		}
		
		if (delayedTasks.areDependenciesFinished()) {
			putAll(delayedTasks.getTasks(), delayedTasks.getPlacement(), false);
		}
	}
	
	synchronized final void onTaskFinished(ITask task) throws InterruptedException {
		task.markExecuted();
		List<DelayedTasks> tasksDepending = tasksWithDependencies.get(task.getUniqueID());
		if (tasksDepending != null) {
			for(DelayedTasks taskDepending : tasksDepending) {
				taskDepending.decrementDependencies();
				if (taskDepending.areDependenciesFinished()) {
					putAll(taskDepending.getTasks(), taskDepending.getPlacement(), true);
				}
			}
		}
	}
	
	public synchronized final void putAll(ITask... newTasks) throws InterruptedException {
		for(ITask newTask : newTasks) {
			put(newTask);
		}
	}
	
	public final void barrier(boolean blocking) throws InterruptedException {
		final int currentNbWorkers = plannedNbWorkers;
			if (currentNbWorkers > 0) {
				final Semaphore workersWaitingForBarrier = new Semaphore(0);
				final Semaphore barrierFinished = new Semaphore(0);
				put(new BarrierTask(workersWaitingForBarrier, barrierFinished, 0));
				
				if (blocking) {
					barrierFinished.acquire();
				}
		}
	}

	public final void releaseWorkers(int releasedWorkers, boolean blocking, TaskPlacement taskPlacement) throws InterruptedException {
		releaseWorkers(releasedWorkers, blocking, taskPlacement, DispatchingEvent.USER_RELEASED_TASK_FEED_WORKER);
	}
	
	private final void releaseWorkers(int releasedWorkers, boolean blocking, TaskPlacement taskPlacement, DispatchingEvent event) throws InterruptedException {
		
		if (releasedWorkers <= 0) {
			throw new IllegalArgumentException("Parameter 'releasedWorkers' must be higher than 0");
		}
		
		if (!closeSignaled && plannedNbWorkers - releasedWorkers < minimalWorkers) {
			throw new IllegalArgumentException("Cannot release " + releasedWorkers +" on a task queue planned to contain " + plannedNbWorkers + " workers, because it must keep at least " + minimalWorkers + " workers");
			
		}
		
		if (plannedNbWorkers < releasedWorkers) {
			throw new IllegalArgumentException("This task queue has less than " + releasedWorkers + " workers");
		}
		Semaphore releasedSemaphore = new Semaphore(0);
//		System.out.println("releasing " + releasedWorkers + " workers, currently " + plannedNbWorkers + " workers" );
		for(int i = 0; i < releasedWorkers; i++ ) {
			
			if (taskPlacement == TaskPlacement.PLACE_AT_START_OF_QUEUE) {
				tasks.putFirst(new ReleaseWorkerTask(false, event, releasedSemaphore));
			} else {
				tasks.putLast(new ReleaseWorkerTask(false, event, releasedSemaphore));
			}
		}
		
		decrementNbWorkers(releasedWorkers);
		
		if(blocking) {
			releasedSemaphore.acquire(releasedWorkers);
		}
	}
	
	public void close(boolean blocking) throws InterruptedException {
		if (!closeSignaled) {
			closeSignaled = true;
			tasks.putLast(new ReleaseWorkerTask(true, null, leaveSemaphore));
		}
		if (blocking) {
			terminatedTaskQueueWorkers.acquire(assignedTaskQueueWorkers);
			//Release them, in case an other thread calls close(true);
			terminatedTaskQueueWorkers.release(assignedTaskQueueWorkers);
		}
		
	}
	
	public Semaphore getTerminatedTaskQueueWorkers() {
		return terminatedTaskQueueWorkers;
	}

	public void cancel(boolean blocking) throws InterruptedException {
		tasks.clear();
		close(blocking);
	}
	
	public int getMinimalWorkers() {
		return minimalWorkers;
	}

	public int getOptimalWorkers() {
		return optimalWorkers == null ? maximalWorkers : optimalWorkers;
	}

	public int getMaximalWorkers() {
		return maximalWorkers;
	}
	
	public int getPlannedNbWorkers() {
		return plannedNbWorkers;
	}
	
	@SuppressWarnings("unchecked")
	public <T> T getSharedParameter(String name) throws InterruptedException {
		modifyingSharedParameter.acquire();
		T value =  ((T) sharedParameters.get(name));
		modifyingSharedParameter.release();
		return value;
	}
	
	public void setSharedParameter(String name, Object value) throws InterruptedException {
		modifyingSharedParameter.acquire();
		sharedParameters.put(name, value);
		modifyingSharedParameter.release();
	}
	
	public void incrementSharedIntegerParameter(String name, int increment) throws InterruptedException {
		modifyingSharedParameter.acquire();
		Object currentValue = sharedParameters.get(name);
		sharedParameters.put(name, currentValue == null ? increment : ((Number)currentValue).intValue() + increment);
		modifyingSharedParameter.release();
	}
	
	public void incrementSharedLongParameter(String name, long increment) throws InterruptedException {
		modifyingSharedParameter.acquire();
		Object currentValue = sharedParameters.get(name);
		sharedParameters.put(name, currentValue == null ? increment : ((Number)currentValue).longValue() + increment);
		modifyingSharedParameter.release();
	}

	public void incrementSharedFloatParameter(String name, float increment) throws InterruptedException {
		modifyingSharedParameter.acquire();
		Object currentValue = sharedParameters.get(name);
		sharedParameters.put(name, currentValue == null ? increment : ((Number)currentValue).floatValue() + increment);
		modifyingSharedParameter.release();
	}

	public void incrementSharedDoubleParameter(String name, double increment) throws InterruptedException {
		modifyingSharedParameter.acquire();
		Object currentValue = sharedParameters.get(name);
		sharedParameters.put(name, currentValue == null ? increment : ((Number)currentValue).doubleValue() + increment);
		modifyingSharedParameter.release();
	}
	
	public void multiplySharedIntegerParameter(String name, int factor) throws InterruptedException {
		modifyingSharedParameter.acquire();
		Object currentValue = sharedParameters.get(name);
		sharedParameters.put(name, currentValue == null ? 0 : ((Number)currentValue).intValue() * factor);
		modifyingSharedParameter.release();
	}
	
	public void multiplySharedLongParameter(String name, long factor) throws InterruptedException {
		modifyingSharedParameter.acquire();
		Object currentValue = sharedParameters.get(name);
		sharedParameters.put(name, currentValue == null ? 0 : ((Number)currentValue).longValue() * factor);
		modifyingSharedParameter.release();
	}

	public void multiplySharedFloatParameter(String name, float factor) throws InterruptedException {
		modifyingSharedParameter.acquire();
		Object currentValue = sharedParameters.get(name);
		sharedParameters.put(name, currentValue == null ? 0 : ((Number)currentValue).floatValue() * factor);
		modifyingSharedParameter.release();
	}

	public void multiplySharedDoubleParameter(String name, double factor) throws InterruptedException {
		modifyingSharedParameter.acquire();
		Object currentValue = sharedParameters.get(name);
		sharedParameters.put(name, currentValue == null ? 0 : ((Number)currentValue).doubleValue() * factor);
		modifyingSharedParameter.release();
	}
	
	public void placeSetSharedParameterTask(String name, Object value, TaskPlacement placement) throws InterruptedException {
		if (placement == TaskPlacement.PLACE_AT_START_OF_QUEUE) {
			putInPriority(new SetParameterTask(name, value));
		} else {
			put(new SetParameterTask(name, value));
		}
	}
	
	public void setMinimalWorkers(int minimalWorkers, TaskPlacement placement) throws InterruptedException {
		if (minimalWorkers > this.minimalWorkers) {
			throw new IllegalArgumentException("It is illegal to set the minimal worker to a higher value");
		}
		if (placement == TaskPlacement.PLACE_AT_START_OF_QUEUE) {
			putInPriority(new SetMinimalWorkersTask(minimalWorkers));
		} else {
			put(new SetMinimalWorkersTask(minimalWorkers));
		}
		
	}

	public void setOptimalWorkers(int optimalWorkers, TaskPlacement placement) throws InterruptedException {
		if (placement == TaskPlacement.PLACE_AT_START_OF_QUEUE) {
			putInPriority(new SetOptimalWorkersTask(optimalWorkers));
		} else {
			put(new SetOptimalWorkersTask(optimalWorkers));
		}
	}
	
	public void setOptimalWorkersAsMaximum(TaskPlacement placement) throws InterruptedException {
		if (placement == TaskPlacement.PLACE_AT_START_OF_QUEUE) {
			putInPriority(new SetOptimalWorkersTask(null));
		} else {
			put(new SetOptimalWorkersTask(null));
		}
	}

	public void setMaximalWorkers(int maximalWorkers, TaskPlacement placement) throws InterruptedException {
		if (placement == TaskPlacement.PLACE_AT_START_OF_QUEUE) {
			putInPriority(new SetMaximalWorkersTask(maximalWorkers));
		} else {
			put(new SetMaximalWorkersTask(maximalWorkers));
		}
	}

	public String getName() {
		return name;
	}
	
	public boolean isClosed() {
		return isCloseSignaled() && currentTotalWorkValue == realisedWorkValue;
	}
	
	public void incrementNbWorkers() {
		plannedNbWorkers ++;
	}
	
	public void decrementNbWorkers() {
		plannedNbWorkers --;
	}
	
	public void decrementNbWorkers(int nb) {
		plannedNbWorkers -= nb;
	}

	public Date getCreation() {
		return creation;
	}

	public synchronized double getRealisedWork() {
		return realisedWorkValue;
	}

	public synchronized double getCurrentTotalWork() {
		return currentTotalWorkValue;
	}

	public synchronized Double getEstimatedWork() {
		return estimatedWorkValue;
	}
	
	public synchronized double getUnrealisedWork() {
		double total = getTotalWorkValue();
		return total - realisedWorkValue;
	}
	
	public int getPriority() {
		return priority;
	}

	public TaskQueue setPriority(int priority) {
		this.priority = priority;
		return this;
	}
	
	public TaskQueue setPriority(Priority priority) {
		this.priority = priority.getPriority();
		return this;
	}
}
